package com.event.ingestion.kafka.streamer;

import com.event.ingestion.kafka.BaseStreamer;

import java.util.ArrayList;
import java.util.List;

/**
 * @author ZzHh
 * @Classname EventAttendeesStream
 * @Description TODO
 * @Date:2020/06/18 11:00
 * @Create by IntelliJ IDEA
 **/

/**
 * EventAttendees数据变换
 */
public class EventAttendeesStream extends BaseStreamer {
    //appid
    @Override
    protected String getApplicationId() {
        return "event-attendees-streamming";
    }

    @Override
    protected String getSourceTopic() {
        return "event_attendees_raw";
    }

    @Override
    protected String getTargetTopic() {
        return "event_attendees";
    }

    //数据解析
    @Override
    protected List<String[]> transform(String [] fields) {
        List<String []> results = new ArrayList<>();
        //event_id
        String event_id = fields[0];

        //stauts -> yes
        if (fields.length > 1 && fields[1] != null) {
            String [] yesUsers = fields[1].split(" ");
            if (yesUsers != null && yesUsers.length > 0) {
                for (String yes : yesUsers) {
                    results.add(new String [] {event_id,yes,"yes"});
                }
            }
        }

        //status -> maybe
        if (fields.length > 2 && fields[2] != null) {
            //split
            String [] maybeUsers = fields[2].split(" ");
            if (maybeUsers != null && maybeUsers.length > 0) {
                for (String maybe : maybeUsers) {
                    results.add(new String [] {event_id,maybe,"maybe"});
                }
            }
        }

        //status -> invite
        if (fields.length > 3 && fields[3] != null) {
            String [] invitedUsers = fields[3].split(" ");
            if (invitedUsers != null && invitedUsers.length > 0) {
                for (String invited : invitedUsers) {
                    results.add(new String [] {event_id,invited,"invited"});
                }
            }
        }

        //status -> no
        if (fields.length > 4 && fields[4] != null) {
            String [] noUsers = fields[4].split(" ");
            if (noUsers != null && noUsers.length > 0) {
                for (String no : noUsers) {
                    results.add(new String [] {event_id,no,"no"});
                }
            }
        }
        return results;
    }

    @Override
    protected Boolean isValid(String[] fields) {
        return (fields.length == 5);
    }

    @Override
    protected Boolean isHeader(String[] fields) {
        //check
        return (isValid(fields) && fields[0].equals("event") && fields[1].equals("yes") && fields[2].equals("maybe"));
    }
}